package drds.server.datasource;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.api_warpper.Lock;
import drds.server.config.Alarms;
import drds.server.config.model.GroupDataSourceConfig;
import drds.server.handler.GetConnectionHandler;
import drds.server.handler.Handler;

/**
 * 必须保证主数据源是OK的。要不然在启动的时候启动不成功，还有在切换的时候如果切换失败需要报警。
 * 
 * @author 曹正辉<330937205@qq.com>
 */
public class GroupDataSource {

	protected static final Logger LOGGER = LoggerFactory.getLogger(GroupDataSource.class);

	public final static ConcurrentLinkedQueue<IConnectionHolder> oldConnectionHolderQueue = new ConcurrentLinkedQueue<IConnectionHolder>();
	/**
	 * 写策略:如果只有一个写库则到唯一的写库,如果没有则直接报错。如果是双主则路由到当前指定进行服务器的主数据库。
	 */
	/**
	 * 读策略:BALANCE
	 */
	/**
	 * 所有SQL都转发到主库
	 */
	public static final int BALANCE_TYPE_ON_WRITE = 0;
	/**
	 * 所有读请求随机分发到wiriterHost对应的readhost执行,writerHost不负担读压力
	 */
	public static final int BALANCE_TYPE_ON_ALL_READ = 3;
	/**
	 * 所有读操作都随机的在writeHost、readhost上分发。
	 */
	public static final int BALANCE_TYPE_ON_WRITE_AND_ALL_READ = 2;
	/**
	 * <pre>
	 * 全部的readHost与stand by writeHost参与select语句的负载均衡，
	 * 简单的说，当双主双从模式(M1->S1，M2->S2，并且M1与M2互为主备)，
	 * 正常情况下，S1,M2,S2都参与select语句的负载均衡。
	 * </pre>
	 */
	public static final int BALANCE_TYPE_NOT_MAIN_MASTER = 1;

	public static final long LONG_TIME = 300000;
	public static final int WEIGHT = 0;

	private final String hostName;

	protected DataSource[] writeDataSources;
	protected Map<Integer, DataSource[]> readDataSourcesMap;

	protected volatile int currentActivedIndex;
	protected volatile boolean initSuccess;

	protected final Lock switchDataBaseLock = new Lock();
	private final List<DataSource> dataSourceList;
	private final int banlanceType;
	private final Random random = new Random();
	private String[] schemas;
	private final GroupDataSourceConfig groupDataSourceConfig;

	public GroupDataSource(String hostName, GroupDataSourceConfig groupDataSourceConfig, DataSource[] writeDataSources, Map<Integer, DataSource[]> readDataSourcesMap, int balanceType) {

		this.hostName = hostName;
		this.groupDataSourceConfig = groupDataSourceConfig;
		this.writeDataSources = writeDataSources;
		this.banlanceType = balanceType;

		Iterator<Map.Entry<Integer, DataSource[]>> iterator = readDataSourcesMap.entrySet().iterator();
		while (iterator.hasNext()) {
			DataSource[] dataSources = iterator.next().getValue();
			if (dataSources.length == 0) {
				iterator.remove();
			}
		}

		this.readDataSourcesMap = readDataSourcesMap;
		this.dataSourceList = this.genDataSourceList();

		setGroupDataSourceProperty();
	}

	public List<DataSource> genDataSourceList() {

		LinkedList<DataSource> dataSourceList = new LinkedList<DataSource>();
		for (DataSource writeDataSource : writeDataSources) {
			if (writeDataSource != null) {
				dataSourceList.add(writeDataSource);
			}
		}

		for (DataSource[] readDataSources : this.readDataSourcesMap.values()) {
			for (DataSource readDataSource : readDataSources) {
				if (readDataSource != null) {
					dataSourceList.add(readDataSource);
				}
			}
		}
		return dataSourceList;
	}

	private void setGroupDataSourceProperty() {
		for (DataSource dataSource : this.dataSourceList) {
			dataSource.setGroupDataSource(this);
		}
	}

	public String getHostName() {
		return hostName;
	}

	public int getCurrentActivedIndex() {
		return currentActivedIndex;
	}

	public boolean isInitSuccess() {
		return initSuccess;
	}

	public DataSource findDataSource(IConnectionHolder connectionHolder) {
		for (DataSource dataSource : this.dataSourceList) {
			/**
			 * 读写性质一样即可
			 */
			if ((dataSource.isReadNode() == connectionHolder.isSlaveDataBase())) {
				return dataSource;
			}
		}
		return null;
	}

	public DataSource[] getWriteDataSources() {
		return writeDataSources;
	}

	public DataSource getWriteDataSource() {
		/** 仅仅是写节点 */
		return writeDataSources[currentActivedIndex];
	}

	public boolean switchSource(int newActivedIndex, boolean isAlarm, String reason) {
		/**
		 * 配置了双主模式下才允许切换数据库
		 */
		if (!checkIndex(newActivedIndex)) {
			return false;
		}

		final Lock switchDataBaseLock = this.switchDataBaseLock;
		switchDataBaseLock.lock();
		try {
			int oldActivedIndex = this.currentActivedIndex;
			/**
			 * currentActivedIndex!=newIndex才进行切换
			 */
			if (oldActivedIndex != newActivedIndex) {
				// switch index
				this.currentActivedIndex = newActivedIndex;

				// init again
				this.init(this.currentActivedIndex);

				// clear all connections
				this.getWriteDataSources()[oldActivedIndex].clearConnectionHolderList("switch datasource");

				return true;
			}
		} finally {
			switchDataBaseLock.unlock();
		}
		return false;
	}

	public void init(int index) {

		if (!checkIndex(index)) {
			index = 0;
		}

		int activeIndex = -1;
		int length = writeDataSources.length;
		for (int i = 0; i < length; i++) {
			int $i = indexFix(i + index);
			if (initDataSource($i, writeDataSources[$i])) {

				/**
				 * 不切换-1时，如果主写挂了 不允许切换过去
				 */
				boolean isNotSwitchDataSource = (groupDataSourceConfig.getSwitchType() == GroupDataSourceConfig.NOT_SWITCH_DATASOURCE);
				if (isNotSwitchDataSource && $i > 0) {
					break;
				}
				activeIndex = $i;
				currentActivedIndex = activeIndex;
				initSuccess = true;
				/**
				 * 如果是WRITE_ONLYONE_NODE策略那么只需要一个数据源即可
				 */
				// if (this.writeType == WRITE_TYPE_ONLYONE) {}

				// only init one write datasource
				Server.getInstance().saveDataHostIndex(hostName, currentActivedIndex);
				break;

			}
		}// end

		if (!checkIndex(activeIndex)) {
			initSuccess = false;
			StringBuilder s = new StringBuilder();
			s.append(Alarms.DEFAULT).append(hostName).append(" init failure");
			LOGGER.error(s.toString());
		}
	}

	private boolean checkIndex(int index) {
		return index >= 0 && index < writeDataSources.length;
	}

	private int indexFix(int index) {
		return index < writeDataSources.length ? index : (index - writeDataSources.length);
	}

	private String getMessage(int index, String info) {
		return new StringBuilder().append(hostName).append(" index:").append(index).append(info).toString();
	}

	private boolean initDataSource(int index, DataSource dataSource) {
		int minSize = dataSource.getMysqlInstanceInfo().getMinSize();

		CopyOnWriteArrayList<IConnectionHolder> connectionHolderList = new CopyOnWriteArrayList<IConnectionHolder>();
		GetConnectionHandler getConnectionHandler = new GetConnectionHandler(connectionHolderList, minSize);
		for (int i = 0; i < minSize; i++) {
			try {
				dataSource.getConnection(this.schemas[i % schemas.length], true, getConnectionHandler, null);
			} catch (Exception e) {
				LOGGER.error(getMessage(index, " init connection error."), e);
			}
		}
		/**
		 * 超时机制
		 */
		long endTime = System.currentTimeMillis() + 60 * 1000;
		// waiting for finish
		while (!getConnectionHandler.finished() && //
				(System.currentTimeMillis() < endTime)) {
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				LOGGER.error("initError", e);
			}
		}
		LOGGER.info("init result :" + getConnectionHandler.getStatusInfo());
		return !connectionHolderList.isEmpty();
	}

	/**
	 * 转移 dataSources 数据库连接到回收区
	 */
	public void shiftConnectionHolderListToOldConnectionHolderQueue() {

		// 清除前一次 reload 转移出去的 old Cons, 避免后端太多的问题
		// can't connect to mysql server ,errmsg:Too many connections
		Iterator<IConnectionHolder> iterator = oldConnectionHolderQueue.iterator();
		while (iterator.hasNext()) {
			IConnectionHolder connectionHolder = iterator.next();
			connectionHolder.close("clear old datasources");
			iterator.remove();
		}

		// 转移本次 old Cons 进入回收区
		for (DataSource dataSource : this.dataSourceList) {
			List<IConnectionHolder> shiftConnectionHolderList = dataSource.shiftConnectionHolderList();
			oldConnectionHolderQueue.addAll(shiftConnectionHolderList);
		}
	}

	/**
	 * 强制清除 dataSources
	 */
	public void clearDataSources(String reason) {

		for (DataSource dataSource : this.dataSourceList) {
			dataSource.clearConnectionHolderList(reason);
		}
	}

	public Collection<DataSource> getDataSourceList() {
		return this.dataSourceList;
	}

	/**
	 * return connection for read balance
	 */
	public void getDataSourceWithBanlanceStrategy(String schema, boolean autocommit, Handler handler, Object attachment, String database) throws Exception {
		DataSource dataSource = null;
		ArrayList<DataSource> dataSourceList = null;
		switch (banlanceType) {
		case BALANCE_TYPE_ON_WRITE:
			dataSource = this.getWriteDataSource();
			break;
		case BALANCE_TYPE_ON_ALL_READ: {
			/**
			 * 不包括写节点
			 */
			dataSourceList = getAllActiveReadWriteDataSourceList(false, false, checkSlaveSynStatus());
			dataSource = randomSelect(dataSourceList);
			break;
		}
		case BALANCE_TYPE_ON_WRITE_AND_ALL_READ: {
			/**
			 * 包括当前的写节点
			 */
			dataSourceList = getAllActiveReadWriteDataSourceList(true, true, checkSlaveSynStatus());
			dataSource = randomSelect(dataSourceList);
			break;
		}
		case BALANCE_TYPE_NOT_MAIN_MASTER: {
			// all read nodes and the standard by masters
			/**
			 * 包括写节点,但不包括当前的写节点。如果前者list为Empty则获取一个写节点
			 */
			dataSourceList = getAllActiveReadWriteDataSourceList(true, false, checkSlaveSynStatus());
			if (dataSourceList.isEmpty()) {
				dataSource = this.getWriteDataSource();
			} else {
				dataSource = randomSelect(dataSourceList);
			}
			break;
		}

		default:
			// return default write data source
			dataSource = this.getWriteDataSource();
		}
		// 统计节点读操作次数
		dataSource.addReadCount();
		dataSource.getConnection(schema, autocommit, handler, attachment);
	}

	/**
	 * slave 读负载均衡，也就是 readSource 之间实现负载均衡
	 */
	public void getConnectionWithBanlanceStrategy(String schema, boolean autocommit, Handler handler, Object attachment, String database) throws Exception {

		ArrayList<DataSource> dataSourceList = getAllActiveReadWriteDataSourceList(false, false, checkSlaveSynStatus());
		DataSource dataSource = randomSelect(dataSourceList);
		// 统计节点读操作次数
		dataSource.addReadCount();
		dataSource.getConnection(schema, autocommit, handler, attachment);
	}

	/**
	 * 从 writeHost 下面的 readHost中随机获取一个 connection, 用于slave注解
	 */
	public boolean getDataSource(String schema, boolean autocommit, Handler handler, Object attachment, String database) throws Exception {
		if (readDataSourcesMap.isEmpty()) {
			LOGGER.error("readhost is empty, readSources is empty.");
			return false;
		}
		int index = Math.abs(random.nextInt(Integer.MAX_VALUE)) % readDataSourcesMap.size();
		DataSource[] readDataSources = this.readDataSourcesMap.get(index);
		if (readDataSources == null) {
			LOGGER.error("readhost is notavailable.");
			return false;
		}

		DataSource dataSource = null;
		for (int i = 0; i < readDataSources.length; i++) {
			dataSource = readDataSources[i];
			if (!isAlive(dataSource)) {
				continue;
			} else {

				if (!checkSlaveSynStatus()) {
					continue;
				} else {

					if (canSelectAsReadNode(dataSource)) {
						break;
					} else {
						continue;
					}

				}

			}
		}
		if (dataSource == null) {
			return false;
		}
		// 统计节点读操作次数
		dataSource.addReadCount();
		dataSource.getConnection(schema, autocommit, handler, attachment);
		return true;

	}

	private boolean checkSlaveSynStatus() {
		return (groupDataSourceConfig.getSlaveThreshold() != -1) && (groupDataSourceConfig.getSwitchType() == GroupDataSourceConfig.SYN_STATUS_SWITCH_DS);
	}

	/**
	 * 
	 * 随机选择，按权重设置随机概率。 在一个截面上碰撞的概率高，但调用量越大分布越均匀，而且按概率使用权重后也比较均匀，有利于动态调整提供者权重。
	 */
	public DataSource randomSelect(ArrayList<DataSource> dataSourceList) {

		if (dataSourceList.isEmpty()) {
			return this.getWriteDataSource();
		} else {

			int size = dataSourceList.size(); // 总个数
			int totalWeight = 0; // 总权重
			boolean sameWeight = true; // 权重是否都一样
			for (int i = 0; i < size; i++) {
				int weight = dataSourceList.get(i).getMysqlInstanceInfo().getWeight();
				totalWeight += weight; // 累计总权重
				if (sameWeight && i > 0 && weight != dataSourceList.get(i - 1).getMysqlInstanceInfo().getWeight()) { // 计算所有权重是否一样
					sameWeight = false;
				}
			}

			if (totalWeight > 0 && !sameWeight) {

				// 如果权重不相同且权重大于0则按总权重数随机
				int offset = random.nextInt(totalWeight);

				// 并确定随机值落在哪个片断上
				for (int i = 0; i < size; i++) {
					offset -= dataSourceList.get(i).getMysqlInstanceInfo().getWeight();
					if (offset < 0) {
						return dataSourceList.get(i);
					}
				}
			}

			// 如果权重相同或权重为0则均等随机
			return dataSourceList.get(random.nextInt(size));
		}
	}

	//
	public int getBalanceType() {
		return banlanceType;
	}

	private boolean isAlive(DataSource dataSource) {
		return true;

	}

	private boolean canSelectAsReadNode(DataSource dataSource) {
		return true;

		/*
		 * Integer slaveBehindMaster =
		 * theSource.getHeartbeat().getSlaveBehindMaster(); int dbSynStatus =
		 * theSource.getHeartbeat().getDbSynStatus();
		 * 
		 * if (slaveBehindMaster == null || dbSynStatus ==
		 * DBHeartbeat.DB_SYN_ERROR) { return false; } boolean isSync =
		 * dbSynStatus == DBHeartbeat.DB_SYN_NORMAL; boolean isNotDelay =
		 * slaveBehindMaster < this.dataHostConfig.getSlaveThreshold(); return
		 * isSync && isNotDelay;
		 */
	}

	/**
	 * return all backup write sources
	 * 
	 * @param includeWriteNode
	 * @param includeCurrentWriteNode
	 * @param filterWithSlaveThreshold
	 *            Threshold[门槛]
	 * 
	 */
	private ArrayList<DataSource> getAllActiveReadWriteDataSourceList(//
			boolean includeWriteNode,//
			boolean includeCurrentWriteNode,//
			boolean filterWithSlaveThreshold) {//

		int currentActiveIndex = this.currentActivedIndex;
		ArrayList<DataSource> dataSourceList = new ArrayList<DataSource>(this.dataSourceList.size());

		for (int i = 0; i < this.writeDataSources.length; i++) {
			DataSource writeDataSource = writeDataSources[i];
			if (isAlive(writeDataSource)) {// write node is active
				if (includeWriteNode) {
					//
					boolean isCurrentWriteNode = (i == currentActiveIndex);
					if (isCurrentWriteNode && includeCurrentWriteNode == false) {
						// not include cur active source
					} else if (filterWithSlaveThreshold && writeDataSource.isSalveOrRead()) {
						boolean selected = canSelectAsReadNode(writeDataSource);
						if (selected) {
							dataSourceList.add(writeDataSource);
						} else {
							continue;
						}
					} else {
						dataSourceList.add(writeDataSource);
					}
				}

				if (!readDataSourcesMap.isEmpty()) {
					// check all slave nodes
					DataSource[] readDataSources = this.readDataSourcesMap.get(i);
					if (readDataSources != null) {
						for (DataSource readDataSource : readDataSources) {
							if (isAlive(readDataSource)) {
								if (filterWithSlaveThreshold) {
									boolean selected = canSelectAsReadNode(readDataSource);
									if (selected) {
										dataSourceList.add(readDataSource);
									} else {
										continue;
									}
								} else {
									dataSourceList.add(readDataSource);
								}
							}
						}
					}
				}

			} else {
				// 当写服务不可用,由都数据库保证临读时服务正确
				if (this.groupDataSourceConfig.isTempReadHostAvailable() && //
						!readDataSourcesMap.isEmpty()) {

					// check all slave nodes
					DataSource[] readDataSources = this.readDataSourcesMap.get(i);
					if (readDataSources != null) {
						for (DataSource readDataSource : readDataSources) {
							if (isAlive(readDataSource)) {

								if (filterWithSlaveThreshold) {
									if (canSelectAsReadNode(readDataSource)) {
										dataSourceList.add(readDataSource);
									} else {
										continue;
									}

								} else {
									dataSourceList.add(readDataSource);
								}
							}
						}
					}
				}
			}

		}// end
		return dataSourceList;
	}

	public String[] getSchemas() {
		return schemas;
	}

	public void setSchemas(String[] schemas) {
		this.schemas = schemas;
	}
}